package main

import (
	"context"
	"fmt"
	"git.inke.cn/BackendPlatform/golang/kafka"
	"git.inke.cn/BackendPlatform/golang/logging"
	"git.inke.cn/inkelogic/daenerys"
	"git.inke.cn/inkelogic/daenerys/proxy"
	"time"
)

const name = "consumer.test"

func main() {
	err := daenerys.Default.InitKafkaConsume([]kafka.KafkaConsumeConfig{
		kafka.KafkaConsumeConfig{
			ConsumeFrom:    name,
			Zookeeperhost:  "10.128.0.104:2181,10.128.0.99:2181,10.128.0.103:2181,10.128.0.102:2181,10.128.0.98:2181/ali-qa/kafka02",
			Topic:          "nvwa_finance_finance_withdrawal_logic_fund_trans_chatright",
			Group:          "cop.inke_owt.nvwa_pdl.test.sc",
			Initoffset:     -2,
			ProcessTimeout: 10,
			CommitInterval: 10,
		},
	})
	if err != nil {
		panic(err)
		return
	}

	CoinRiskConsumer(context.Background(), name)
	time.Sleep(time.Hour * 2)
}

func CoinRiskConsumer(ctx context.Context, consumerName string) {
	nameSpace := daenerys.GetAPPKey(ctx)

	logging.Infof("CoinRiskConsumer|nameSpace: %s|Start WithDrawalStatusConsumer ...", nameSpace)

	consumer := proxy.InitKafkaConsumer(consumerName)

	msgCh := consumer.GetMessages(ctx)
	if msgCh == nil {
		logging.Warningf("CoinRiskConsumer| msgCh %+v", msgCh)
	}
	for {
		msg, ok := <-msgCh
		if !ok {
			logging.Warning("CoinRiskConsumer| msgCh closed")
			return
		}
		fmt.Printf("msg=%s,partition=%d", string(msg.Value), msg.Partition)
		msgCtx := daenerys.WithAPPKey(msg.Context(), nameSpace)
		if err := coinRiskHandler(msgCtx, msg); err == nil {
			consumer.CommitUpto(ctx, msg)
		}
	}

}

func coinRiskHandler(ctx context.Context, msg *kafka.ConsumerMessage) error {
	time.Sleep(time.Millisecond * 40)
	return nil
}
